package handle

import (
	"fmt"
	"log"

	"github.com/rabbitmq/amqp091-go"
)

type RabbitConn struct {
	conn      *amqp091.Connection
	channel   *amqp091.Channel
	ConfigUrl string

	tag string
}

func (r *RabbitConn) NotifyCloseGo() {
	fmt.Printf("closing: %s", <-r.conn.NotifyClose(make(chan *amqp091.Error)))
}

func (r *RabbitConn) QueueBind(routingKey, exchange string, queue string) {
	if err := r.channel.QueueBind(
		queue,      // name of the queue
		routingKey, // routingKey
		exchange,   // sourceExchange
		false,      // noWait
		nil,        // arguments
	); err != nil {
		log.Fatalf("QueueBind: %s", err)
	}
}

func (r *RabbitConn) Consume(queue string, args amqp091.Table, handle func(message amqp091.Delivery)) {
	deliveries, err := r.channel.Consume(
		queue, // name
		r.tag, // consumerTag,
		false, // noAck
		false, // exclusive
		false, // noLocal
		false, // noWait
		args,  // arguments
	)
	if err != nil {
		panic(err)
	}
	r.handleFun(deliveries, handle)
}
func (r *RabbitConn) Shutdown() error {

	// will close() the deliveries channel
	if err := r.channel.Cancel(r.tag, true); err != nil {
		return fmt.Errorf("RabbitConn cancel failed: %s", err)
	}

	if err := r.conn.Close(); err != nil {
		return fmt.Errorf("AMQP connection close error: %s", err)
	}

	defer log.Printf("AMQP shutdown OK")

	// wait for handle() to exit
	return nil
}

func (r *RabbitConn) handleFun(deliveries <-chan amqp091.Delivery, handle func(message amqp091.Delivery)) {
	for d := range deliveries {
		handle(d)
	}
	log.Println("handleFun: deliveries channel closed")
}
